package com.biz.primus.ms.stock.service;


import com.biz.primus.base.exception.utils.AssertUtils;
import com.biz.primus.common.enums.StockChannelEnum;
import com.biz.primus.common.utils.JsonUtils;
import com.biz.primus.model.stock.enums.StockChangeTypeEnum;
import com.biz.primus.model.stock.exceptions.StockException;
import com.biz.primus.model.stock.vo.*;
import com.biz.primus.ms.base.service.AbstractBaseService;
import com.biz.primus.ms.stock.component.annotation.StockChangMark;
import com.biz.primus.ms.stock.dao.redis.DepotStockRedisDao;
import com.biz.primus.ms.stock.dao.ro.Constants;
import com.biz.primus.ms.stock.dao.ro.DepotStockRo;
import com.biz.primus.ms.stock.feign.DepotFeignClient;
import com.biz.primus.ms.stock.stream.StockChangeSource;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * AbstractStockService
 * 库存基本服务
 *
 * @author JKLiues
 * @date 2017/10/20
 */
@Slf4j
public abstract class AbstractStockService extends AbstractBaseService {

    @Autowired
    protected DepotStockRedisDao depotStockRedisDao;

    @Autowired
    protected DepotFeignClient depotFeignClient;

    @Autowired
    protected StockChangeSource stockChangeSource;

    /**
     * 写全量库存
     *
     * @param reqVos 全量库存子项明细集合
     */
    public void saveDepotStocks(List<AllStockItemVo> reqVos) {
        List<AllStockItemVo> validStockVos = this.validStockVos(reqVos);
        if (CollectionUtils.isEmpty(validStockVos)) {
            return;
        }
        List<StockChannelEnum> collect = reqVos.stream().map(AllStockItemVo::getChannelCode).collect(Collectors.toList());
        Map<StockChannelEnum, Long> versionMap = collect.stream()
                .collect(Collectors.toMap(channelCode -> channelCode, channelCode -> depotStockRedisDao.getVersion(channelCode) + 1L));
        List<DepotStockRo> depotStockRos = validStockVos.stream()
                .map(vo -> new DepotStockRo(vo, versionMap.get(vo.getChannelCode())))
                .collect(Collectors.toList());
        depotStockRedisDao.save(depotStockRos);
        log.info("处理全量库存,{}/{}", reqVos.size(), depotStockRos.size());
    }

    /**
     * 全量库存后置操作
     *
     * @param channelCode 渠道码
     */
    public abstract void afterHandleStockAll(StockChannelEnum channelCode);

    /**
     * 增量库存
     * 常见于线下库存推送
     *
     * @param reqVo 增量库存对象
     */
    @StockChangMark
    public void updateStockOnly(IncrDepotStockReqVo reqVo) {
        validateIncrDepotStocks(reqVo);
        String depotCode = reqVo.getDepotCode();
        StockChannelEnum channelCode = StockChannelEnum.NORMAL;
        List<ProductQuantityPair> items = reqVo.getItems();
        log.debug("更新门店库存{}", reqVo);
        Set<String> productCodes = items.stream().map(ProductQuantityPair::getProductCode).collect(Collectors.toSet());
        Map<String, Integer> idToQuantity = items.stream().collect(Collectors.toMap(pair -> {
            String productCode = pair.getProductCode();
            return String.format("%s:%s:%s", channelCode, depotCode, productCode);
        }, ProductQuantityPair::getQuantity));
        depotStockRedisDao.updateDepotStockQuantityInBatch(idToQuantity);
        this.pushStockChangeMessage(depotCode, productCodes);
    }

    /**
     * 门店锁库存更新
     * 常见于线上下单未支付，订单过期释放
     *
     * @param reqVo 锁库存对象
     */
    public void updateLockedStockOnly(IncrDepotStockReqVo reqVo) {
        validateIncrDepotStocks(reqVo);
        StockChannelEnum channelCode = StockChannelEnum.NORMAL;
        List<ProductQuantityPair> itemVos = reqVo.getItems();
        String depotCode = reqVo.getDepotCode();
        Set<String> productCodes = itemVos.stream().map(ProductQuantityPair::getProductCode).collect(Collectors.toSet());
        Map<String, Integer> idToLockedQuantity = itemVos.stream()
                .collect(Collectors.toMap(
                        vo -> String.format("%s:%s:%s", channelCode, depotCode, vo.getProductCode()),
                        ProductQuantityPair::getQuantity));
        depotStockRedisDao.updateDepotLockedStockQuantityInBatch(idToLockedQuantity);
        this.pushStockChangeMessage(depotCode, productCodes);
    }

    /**
     * 更新库存同时更新锁库存
     * 常见于线上订单支付完成扣减锁库存
     *
     * @param reqVo 库存更新对象
     */
    @StockChangMark
    public void updateStockAndLockedStock(IncrDepotStockReqVo reqVo) {
        validateIncrDepotStocks(reqVo);
        List<ProductQuantityPair> itemVos = reqVo.getItems();
        String depotCode = reqVo.getDepotCode();
        List<String> productCodes = itemVos.stream()
                .map(ProductQuantityPair::getProductCode)
                .collect(Collectors.toList());
        Map<String, Integer> idToQuantityMap = itemVos.stream()
                .collect(Collectors.toMap(vo -> String.format("%s%s%s", depotCode, Constants.SEPARATOR, vo.getProductCode()), ProductQuantityPair::getQuantity));
        depotStockRedisDao.updateDepotStockQuantityInBatch(idToQuantityMap);
        depotStockRedisDao.updateDepotLockedStockQuantityInBatch(idToQuantityMap);
        this.pushStockChangeMessage(depotCode, productCodes);
    }

    /**
     * 校验从中台过来的全量库存对象中的参数
     *
     * @param vo 校验对象
     * @return boolean
     */
    private Boolean validateVos(AllStockItemVo vo) {
        if (StringUtils.isBlank(vo.getDepotCode()) ||
                StringUtils.isBlank(vo.getProductCode())) {
            log.debug("未通过校验Vo序列化为: {}", JsonUtils.obj2Json(vo));
            return false;
        }
        return true;
    }

    /**
     * 校验库存和锁库存请求对象
     *
     * @param reqVo 请求对象
     */
    private void validateIncrDepotStocks(IncrDepotStockReqVo reqVo) {
        String depotCode = reqVo.getDepotCode();
        AssertUtils.hasLength(depotCode, StockException.DEPOT_CODE_IS_NULL);
        String bn = reqVo.getBn();
        AssertUtils.hasLength(bn, StockException.BN_IS_NULL);
        StockChangeTypeEnum changeType = reqVo.getChangeType();
        AssertUtils.notNull(changeType, StockException.STOCK_CHANGE_TYPE_IS_NULL);
        List<ProductQuantityPair> items = reqVo.getItems();
        AssertUtils.notEmpty(items, StockException.CHANGE_STOCK_IS_EMPTY);
        for (ProductQuantityPair pair : items) {
            AssertUtils.hasLength(pair.getProductCode(), StockException.PRODUCT_CODE_IS_NULL);
            AssertUtils.notNull(pair.getQuantity(), StockException.QUANTITY_IS_NULL);
        }
        Boolean bnEffective = depotStockRedisDao.isBnEffective(bn, changeType);
        AssertUtils.isTrue(bnEffective, StockException.BN_REPEATED);
    }

    /**
     * 校验中台全量库存
     *
     * @param reqVos 全量库存
     * @return List<AllDepotStockReqVo>
     */
    private List<AllStockItemVo> validStockVos(List<AllStockItemVo> reqVos) {
        if (CollectionUtils.isEmpty(reqVos)) {
            log.warn("中台全量库存Vo集合为空");
            return Lists.newArrayList();
        }
        return reqVos.stream().filter(this::validateVos).collect(Collectors.toList());

    }

    private void pushStockChangeMessage(String depotCode, Collection<String> productCode) {
        StockChangeMessageVo vo = new StockChangeMessageVo();
        vo.setDepotCode(depotCode);
        vo.setProductCode(Sets.newHashSet(productCode));
        stockChangeSource.change().send(MessageBuilder.withPayload(vo).build());
    }

    protected void pushStockChangeAllMessage() {
        stockChangeSource.changeAll().send(MessageBuilder.withPayload(new StockChangeAllMessageVo()).build());
    }
}
